package pipeline

import (
	"fmt"
	"sync"
	"time"
)

const (
	defaultBufferSize = 10
)

// nodes

func Buy(n int) <-chan string {
	ch := make(chan string, defaultBufferSize)
	go func() {
		for i := 0; i < n; i++ {
			ch <- fmt.Sprintf("购买零件-%d", i)
			// 100ms 处理一个
			time.Sleep(time.Millisecond * 100)
		}
		// 处理n个数据项, stream 有明显结束标志
		close(ch)
	}()
	return ch
}

func Build(in <-chan string) <-chan string {
	out := make(chan string, defaultBufferSize)
	go func() {
		for val := range in {
			out <- fmt.Sprintf("组装(%s)", val)
			// 300ms 处理一个
			time.Sleep(time.Millisecond * 300)
		}
		close(out)
	}()
	return out
}

func Pack(in <-chan string) <-chan string {
	out := make(chan string, defaultBufferSize)
	go func() {
		for val := range in {
			out <- fmt.Sprintf("打包(%s)", val)
			// 100ms 处理一个
			time.Sleep(time.Millisecond * 200)
		}
		close(out)
	}()
	return out
}

// merge

func Merge(ins ...<-chan string) <-chan string {
	var wg sync.WaitGroup
	wg.Add(len(ins))
	out := make(chan string, defaultBufferSize*len(ins))
	go func() {
		for _, in := range ins {
			go func(in <-chan string) {
				for val := range in {
					out <- val
				}
				wg.Done()
			}(in)
		}
		// 等待所有上游节点关闭, merge 才关闭
		wg.Wait()
		close(out)
	}()
	return out
}
